package com.yvan.eventsourcing;

import com.google.common.collect.Maps;
import com.rabbitmq.client.Channel;
import com.yvan.Conv;
import com.yvan.YvanUtil;
import com.yvan.platform.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.beans.factory.InitializingBean;

import java.util.Map;

/**
 * 支持重试的 amqp 封装
 * Created by luoyifan on 2017/8/15.
 */
public class EventContainerFactoryBean<T> implements InitializingBean, ChannelAwareMessageListener {

    private static final Logger LOGGER = LoggerFactory.getLogger(com.yvan.eventsourcing.EventContainerFactoryBean.class);
    private static final String RETRY_ROUTING_KEY = "retry";
    private static final String RETRY_MESSAGE_PROP_KEY = "retry_count";
    private static final String VERSION_PROP_KEY = "version";

    private volatile MessageConverter messageConverter = new SimpleMessageConverter();

    private volatile boolean autoStart = true;
    private volatile String exchange;
    private volatile String errorExchange;
    private volatile String routingKey;
    private volatile String queue;
    private volatile int concurrentConsumers = 1;
    private volatile RetryStrategy retryStrategy = RetryStrategy.DEFAULT;
    private volatile String version = "";
    private volatile ConnectionFactory connectionFactory;
    private volatile EventHandler<T> eventHandler;
    private volatile SimpleMessageListenerContainer listenerContainer;
    private volatile RabbitTemplate rabbitTemplate;

    protected EventContainerFactoryBean() {

    }

    @Override
    public void afterPropertiesSet() throws Exception {
        init();
        LOGGER.info("retry queue:" + exchange + " inited.");

        listenerContainer = new SimpleMessageListenerContainer(connectionFactory);
        listenerContainer.setExposeListenerChannel(true);
        listenerContainer.setMessageConverter(messageConverter);
        listenerContainer.setQueueNames(queue);
        listenerContainer.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        listenerContainer.setMessageListener(this);
        listenerContainer.setConcurrentConsumers(concurrentConsumers);
        listenerContainer.setMaxConcurrentConsumers(concurrentConsumers);

        if (rabbitTemplate == null) {
            rabbitTemplate = new RabbitTemplate(connectionFactory);
            rabbitTemplate.setMessageConverter(messageConverter);
        }

        if (autoStart) {
            start();
        }
    }

    @SuppressWarnings("unchecked")
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long tag = message.getMessageProperties().getDeliveryTag();
        int retryCount = Conv.NI(message.getMessageProperties().getHeaders().getOrDefault(RETRY_MESSAGE_PROP_KEY, "0"));
        String messageVersion = Conv.NS(message.getMessageProperties().getHeaders().getOrDefault(VERSION_PROP_KEY, ""));
        EventHandler.Action action = EventHandler.Action.RETRY;
        Object object = null;
        String objectJson = "null";

        try {
            object = messageConverter.fromMessage(message);
            if (version == null || StringUtils.isNullOrEmpty(messageVersion) || version.equals(messageVersion)) {
                action = eventHandler.handle((T) object);
                objectJson = YvanUtil.toJson(object);
            } else {
                action = EventHandler.Action.RETRY;
                LOGGER.info("version not match,will be putting into retry");
            }
        } catch (Exception e) {
            action = EventHandler.Action.RETRY;
            LOGGER.error("[" + exchange + "].onMessage error, object:" + objectJson, e);
            throw e;

        } finally {
            if (action == EventHandler.Action.RETRY) {
                MessageProperties properties = new MessageProperties();
                if (object == null) {
                    LOGGER.info("retry fail, object is null");

                } else {
                    retryCount++;
                    LOGGER.info("retry [" + exchange + "], object:" + objectJson);

                    properties.getHeaders().put(RETRY_MESSAGE_PROP_KEY, retryCount);
                    properties.getHeaders().put(VERSION_PROP_KEY, messageVersion);
                    long ttl = retryStrategy.getTtl(retryCount);
                    if (ttl == -1) {
                        LOGGER.info("retry end [" + exchange + "], object:" + objectJson + ", retry count:" + retryCount);
                    } else {
                        properties.setExpiration(Conv.NS(retryStrategy.getTtl(retryCount)));
                        message = messageConverter.toMessage(object, properties);
                        rabbitTemplate.send(getErrorExchange(), getQueue(), message);
                    }
                }

            } else if (action == EventHandler.Action.REJECT) {
                LOGGER.warn("reject [" + exchange + "], object:" + objectJson);

            }

            if (action == EventHandler.Action.PASS) {
                channel.basicRecover(true);
            } else {
                channel.basicAck(tag, false);
            }
        }
    }

    private void init() {
        RabbitAdmin admin = new RabbitAdmin(connectionFactory);

        admin.declareExchange(new DirectExchange(getExchange()));
        admin.declareExchange(new DirectExchange(getErrorExchange()));
        admin.declareQueue(new Queue(getQueue()));
        admin.declareBinding(new Binding(getQueue(), Binding.DestinationType.QUEUE, getExchange(), getRoutingKey(), null));

        //make dlx
        Map<String, Object> args = Maps.newHashMap();
        args.put("x-dead-letter-exchange", getErrorExchange());
        args.put("x-dead-letter-routing-key", getRetryRouteKey());
        admin.declareQueue(new Queue(getDeadLetterQueueName(), true, false, false, args));
        admin.declareBinding(new Binding(getDeadLetterQueueName(), Binding.DestinationType.QUEUE, getErrorExchange(), getQueue(), null));
        admin.declareBinding(new Binding(getQueue(), Binding.DestinationType.QUEUE, getErrorExchange(), getRetryRouteKey(), null));
    }

    private String getRetryRouteKey() {
        return RETRY_ROUTING_KEY + "_" + getQueue();
    }

    private String getDeadLetterQueueName() {
        return getQueue() + "_retry";
    }

    public void start() {
        listenerContainer.start();
    }

    public void stop() {
        listenerContainer.stop();
    }

    public MessageConverter getMessageConverter() {
        return messageConverter;
    }

    public void setMessageConverter(MessageConverter messageConverter) {
        this.messageConverter = messageConverter;
    }

    public String getExchange() {
        return exchange;
    }

    public void setExchange(String exchange) {
        this.exchange = exchange;
    }

    public String getErrorExchange() {
        return errorExchange;
    }

    public void setErrorExchange(String errorExchange) {
        this.errorExchange = errorExchange;
    }

    public String getRoutingKey() {
        return routingKey;
    }

    public void setRoutingKey(String routingKey) {
        this.routingKey = routingKey;
    }

    public String getQueue() {
        return queue;
    }

    public void setQueue(String queue) {
        this.queue = queue;
    }

    public int getConcurrentConsumers() {
        return concurrentConsumers;
    }

    public void setConcurrentConsumers(int concurrentConsumers) {
        this.concurrentConsumers = concurrentConsumers;
    }

    public RetryStrategy getRetryStrategy() {
        return retryStrategy;
    }

    public void setRetryStrategy(RetryStrategy retryStrategy) {
        this.retryStrategy = retryStrategy;
    }

    public ConnectionFactory getConnectionFactory() {
        return connectionFactory;
    }

    public void setConnectionFactory(ConnectionFactory connectionFactory) {
        this.connectionFactory = connectionFactory;
    }

    public EventHandler<T> getEventHandler() {
        return eventHandler;
    }

    public void setEventHandler(EventHandler<T> eventHandler) {
        this.eventHandler = eventHandler;
    }

    public boolean isAutoStart() {
        return autoStart;
    }

    public void setAutoStart(boolean autoStart) {
        this.autoStart = autoStart;
    }

    public RabbitTemplate getRabbitTemplate() {
        return rabbitTemplate;
    }

    public void setRabbitTemplate(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    public String getVersion() {
        return version;
    }

    public void setVersion(String version) {
        this.version = version;
    }
}
